package pipe

import "sync"

/**
                    ——> method:1
                   |
db -> chan ----channel ----> han ---> result
                   |
                   ——> method:n
使用channel解决并发

*/

type C chan interface{}
type InFunc func(args ...interface{}) C
type PipeFunc func(in C) C

type Pipe struct {
	In    InFunc
	Pipe  PipeFunc
	Count int // 多路复用需要的参数
}

func NewPipe() *Pipe {
	// 初始化：单路
	return &Pipe{Count: 1}
}

func (this *Pipe) SetIn(in InFunc) *Pipe {
	this.In = in
	return this
}

func (this *Pipe) SetPipe(p PipeFunc, count int) *Pipe {
	this.Pipe = p
	this.Count = count
	return this
}

// 实现多路复用
func (this *Pipe) Exec(args ...interface{}) C {
	in := this.In(args)
	out := make(C)
	wg := sync.WaitGroup{}
	for i := 0; i < this.Count; i++ {
		c := this.Pipe(in)
		wg.Add(1)
		go func(c C) {
			defer wg.Done()
			for v := range c {
				out <- v
			}
		}(c)
	}
	go func() {
		defer close(out)
		wg.Wait()
	}()
	return out
}
